#include "stdafx.h"


#include "Socket.h"
#include "PC2DataStore.h"

#include "PC2Coordinator.h"


namespace PC2
{

PC2Coordinator::PC2Coordinator( NodeInfo const& aID, std::vector<NodeInfo> const& aParticipantIDSet, char const* apDBPath )
	: PC2Node(aID, aParticipantIDSet, apDBPath)
	, m_pExecutor(NULL)
{
}

PC2Coordinator::~PC2Coordinator()
{
}


void PC2Coordinator::SetExecutor(ManagerExecutor* apExecutor)
{
	m_pExecutor = apExecutor;
}

int32_t PC2Coordinator::Recovery(std::string* apData)
{
	TransactionInfoSet lSet;
	uint32_t lCurrentTransID = m_pStore->GetCurrentTransId();
	if(0 == lCurrentTransID)
	{
		return 0;
	}
	else
	{
		m_pStore->LoadTranInfoSet(lCurrentTransID, &lSet, true);
		if(EnumCommitAck == lSet.m_LastTrans.m_Type || EnumAbortAck == lSet.m_LastTrans.m_Type)
		{
			return 0;
		}
	}
	
	if(PerformStoredAction(&lSet))
	{
		return 1;
	}
	else
	{
		return -1;
	}
}


void PC2Coordinator::WaitAndRunOnce(uint32_t aTimeout)
{
	while (HandleIncomingMessage(NULL, aTimeout))
	{
	}
}

bool PC2Coordinator::PerformNewAction( char const* apData, uint32_t aSize )
{
	TransactionInfoSet lSet;
	InitAndStorePrepareRequest(&lSet, apData, aSize);
	return PerformStoredAction(&lSet);
}

bool PC2Coordinator::PerformStoredAction(TransactionInfoSet* apInfoSet)
{
	if(EnumPrepareRequest == apInfoSet->m_LastTrans.m_Type)
	{
		LOG("Do phrase commit 1 trans_id %u\n", apInfoSet->m_LastTrans.m_TransId);

		LOG("receiving for prepare resp\n");
		SendRequestForCurrentPhase(apInfoSet);
		while(apInfoSet->m_PrepareResponses.size() < apInfoSet->m_RelatedNodes.size())
		{	
			bool lReceived = HandleIncomingMessage(apInfoSet, _TIMEOUT_MS);
			if(false == lReceived)
				break;
		}

		if(apInfoSet->m_PrepareResponses.size() == apInfoSet->m_RelatedNodes.size()
			&& JudgePrepareResult(apInfoSet))
		{
			LOG("all vote YES\n");
			apInfoSet->m_LastTrans.m_Type = EnumCommitRequest;
		}
		else
		{
			LOG("someone vote NO or wait timeout\n");
			apInfoSet->m_LastTrans.m_Type = EnumAbortRequest;
		}
		apInfoSet->m_LastTrans.m_Size = sizeof(Trans);
		leveldb::WriteBatch lBatch;
		PC2DataStore::WriteLastTransState(&lBatch, apInfoSet);
		m_pStore->CommitBatch(&lBatch);

		LOG("phrase commit 1 trans_id %u done\n", apInfoSet->m_LastTrans.m_TransId);
	}

	if(EnumCommitRequest == apInfoSet->m_LastTrans.m_Type || EnumAbortRequest == apInfoSet->m_LastTrans.m_Type)
	{
		LOG("Do phrase commit 2 trans_id %u\n", apInfoSet->m_LastTrans.m_TransId);

		LOG("send action for nodes which vote yes\n");
		SendRequestForCurrentPhase(apInfoSet);

		bool lOK;
		if(EnumCommitRequest == apInfoSet->m_LastTrans.m_Type)
		{
			apInfoSet->m_LastTrans.m_Type = EnumCommitAck;
			lOK = true;
		}
		else
		{
			apInfoSet->m_LastTrans.m_Type = EnumAbortAck;
			lOK = false;
		}
		apInfoSet->m_LastTrans.m_Size = sizeof(Trans);
		
		leveldb::WriteBatch lBatch;
		PC2DataStore::WriteLastTransState(&lBatch, apInfoSet);
		PC2DataStore::FinishAndClearTransactionInfo(&lBatch, apInfoSet);
		m_pStore->CommitBatch(&lBatch);

		LOG("phrase commit 2 trans_id %u done\n", apInfoSet->m_LastTrans.m_TransId);
		return lOK;
	}
	else
	{
		LOG("unknown trans state\n");
		return false;
	}
}

bool PC2Coordinator::HandleIncomingMessage( TransactionInfoSet* apInfoSet, uint32_t aTimeout )
{
	char lRemoteIP[32];
	unsigned short lRemotePort;

	
	Trans lReceivedTrans;
	std::string lData;
		
	bool lRecevied = WaitForData(&lReceivedTrans, &lData, aTimeout, lRemoteIP, &lRemotePort);

	if(false == lRecevied)
		return false;

	if(lReceivedTrans.m_Type == EnumPrepareOKResponse || lReceivedTrans.m_Type == EnumPrepareRejectResponse)
	{
		LOG("received prepare resp from node_id %u\n", lReceivedTrans.m_NodeId);
		HandlePrepareResponse(apInfoSet, &lReceivedTrans);
	}
	else if(lReceivedTrans.m_Type == EnumDecisionRequest)
	{
		LOG("received decision req from trans_id %u node_id %u\n", lReceivedTrans.m_TransId, lReceivedTrans.m_NodeId);
		HandleDecisionRequest(apInfoSet, &lReceivedTrans, lRemoteIP, lRemotePort);
	}
	else if(EnumEcho == lReceivedTrans.m_Type)
	{
		LOG("received echo\n");
		NodeInfo lNodeInfo;
		lNodeInfo.m_IP = lRemoteIP;
		lNodeInfo.m_Port = lRemotePort;
		HandleEcho(&lNodeInfo, &lReceivedTrans);
	}
	else if(EnumPrepareRequest == lReceivedTrans.m_Type)
	{
		LOG("received client request\n");
		NodeInfo lNodeInfo;
		lNodeInfo.m_IP = lRemoteIP;
		lNodeInfo.m_Port = lRemotePort;
		bool lISBusy = (NULL != apInfoSet);
		HandleClientRequest(&lNodeInfo, lData.c_str(), lData.length(), lISBusy);
	}

	return true;
}


void PC2Coordinator::HandleEcho(NodeInfo* apNodeInfo, Trans* apReceivedTrans)
{
	Trans lSendTrans;
	LOG("received echo, echo back\n");
	lSendTrans.m_TransId = 0;
	lSendTrans.m_NodeId = m_Info.m_Id;
	lSendTrans.m_Size = sizeof(Trans);
	lSendTrans.m_Type = EnumEcho;
	SendTrans(apNodeInfo, &lSendTrans);
}


void PC2Coordinator::HandleClientRequest(NodeInfo* apNodeInfo, char const* apData, uint32_t aSize, bool aIsBusy)
{
	Trans lSendTrans;
	lSendTrans.m_TransId = 0;
	lSendTrans.m_NodeId = m_Info.m_Id;
	lSendTrans.m_Size = sizeof(Trans);

	if(aIsBusy)
	{
		LOG("new client request coming while busy, rejected\n");
		lSendTrans.m_Type = EnumAbortRequest;
		SendTrans(apNodeInfo, &lSendTrans);
		return;
	}
	
	if(false == m_pExecutor->PrepareClientRequest(apData, aSize))
	{
		lSendTrans.m_Type = EnumAbortRequest;
		SendTrans(apNodeInfo, &lSendTrans);
		return;
	}

	if(PerformNewAction(apData, aSize))
	{
		LOG("transaction OK\n");
		lSendTrans.m_Type = EnumCommitRequest;
		SendTrans(apNodeInfo, &lSendTrans);
	}
	else
	{
		printf("transaction Failed \n");
		lSendTrans.m_Type = EnumAbortRequest;
		SendTrans(apNodeInfo, &lSendTrans);
	}

}

void PC2Coordinator::HandlePrepareResponse(TransactionInfoSet* apInfoSet, Trans* apReceivedTrans)
{
	if(NULL == apInfoSet)
	{
		LOG("currently not in waiting, ignore\n");
		return;
	}

	if(apReceivedTrans->m_TransId != apInfoSet->m_LastTrans.m_TransId)
	{
		LOG("received different trans_id node_id %u trans_id %u type %u, expect node_id %u\n"
			, apReceivedTrans->m_NodeId, apReceivedTrans->m_TransId, apReceivedTrans->m_Type, apInfoSet->m_LastTrans.m_TransId);
		return;
	}

	if(apInfoSet->m_RelatedNodes.end() == apInfoSet->m_RelatedNodes.find(apReceivedTrans->m_NodeId))
	{
		LOG("received from not related node, node_id %u trans_id %u type %u, expect node_id %u\n"
			, apReceivedTrans->m_NodeId, apReceivedTrans->m_TransId, apReceivedTrans->m_Type, apInfoSet->m_LastTrans.m_TransId);
		return;
	}

	if(EnumPrepareRequest != apInfoSet->m_LastTrans.m_Type)
	{
		LOG("received prepare response but too late, node_id %u trans_id %u type %u, expect node_id %u\n"
			, apReceivedTrans->m_NodeId, apReceivedTrans->m_TransId, apReceivedTrans->m_Type, apInfoSet->m_LastTrans.m_TransId);
		return;
	}

	if(apInfoSet->m_PrepareResponses.end() != apInfoSet->m_PrepareResponses.find(apReceivedTrans->m_NodeId))
	{
		return;
	}

	apInfoSet->m_PrepareResponses.insert(std::make_pair(apReceivedTrans->m_NodeId, *apReceivedTrans));
	leveldb::WriteBatch lBatch;
	PC2DataStore::WritePrepareResponseSet(&lBatch, apInfoSet);
	m_pStore->CommitBatch(&lBatch);
}

bool PC2Coordinator::JudgePrepareResult( TransactionInfoSet* apInfoSet )
{
	if(apInfoSet->m_PrepareResponses.size() < apInfoSet->m_RelatedNodes.size())
	{
		return false;
	}

	for (auto it = apInfoSet->m_PrepareResponses.begin(); it != apInfoSet->m_PrepareResponses.end(); it++)
	{
		if(EnumPrepareRejectResponse == it->second.m_Type)
		{
			return false;
		}
	}
	return true;
}

void PC2Coordinator::SendRequestForCurrentPhase( TransactionInfoSet* apInfoSet )
{
	char lBuffer[1024];
	Trans* lpResendTrans = (Trans*)lBuffer;
	lpResendTrans->m_Type = apInfoSet->m_LastTrans.m_Type;
	lpResendTrans->m_NodeId = m_Info.m_Id;
	lpResendTrans->m_TransId = apInfoSet->m_LastTrans.m_TransId;
	
	if( EnumPrepareRequest == lpResendTrans->m_Type)
	{
		NodesHeader* lpNodesHeader = (NodesHeader*)(lpResendTrans + 1);
		lpNodesHeader->WriteParts(&apInfoSet->m_RelatedNodes);

		char* lpDest = (char*)(lpNodesHeader) + lpNodesHeader->m_Size;
		memcpy(lpDest, apInfoSet->m_TransData.c_str(), apInfoSet->m_TransData.length());
		lpResendTrans->m_Size = sizeof(Trans) + lpNodesHeader->m_Size + apInfoSet->m_TransData.length();
	}
	else
	{
		lpResendTrans->m_Size = sizeof(Trans);
	}

	std::set<uint32_t> lNotReceivedSet = FilterNotReceivedForCurrentPhase(apInfoSet);

	if(lNotReceivedSet.empty())
	{
		return;
	}

	for(auto it = lNotReceivedSet.begin();it != lNotReceivedSet.end(); it++)
	{
		NodeInfo* lpPartInfo = &m_Parts.at(*it);
		SendTrans(lpPartInfo, lpResendTrans);
	}
}

std::set<uint32_t> PC2Coordinator::FilterNotReceivedForCurrentPhase( TransactionInfoSet* apInfoSet )
{
	std::set<uint32_t> lNotReceivedSet;
	if(EnumPrepareRequest == apInfoSet->m_LastTrans.m_Type)
	{
		for (auto it = apInfoSet->m_RelatedNodes.begin(); it != apInfoSet->m_RelatedNodes.end(); it++)
		{
			lNotReceivedSet.insert(*it);
		}
		for (auto it = apInfoSet->m_PrepareResponses.begin();it != apInfoSet->m_PrepareResponses.end();it++)
		{
			lNotReceivedSet.erase(it->first);
		}
	}
	else
	{
		for (auto it = apInfoSet->m_PrepareResponses.begin(); it != apInfoSet->m_PrepareResponses.end(); it++)
		{
			if(EnumPrepareOKResponse == it->second.m_Type)
			{
				lNotReceivedSet.insert(it->first);
			}
		}
	}

	return lNotReceivedSet;
}

void PC2Coordinator::InitAndStorePrepareRequest( TransactionInfoSet* apInfoSet, char const* apData, uint32_t aSize )
{
	uint32_t lTransId = m_pStore->GetCurrentTransId();
	if(0 != lTransId)
	{
		lTransId += 1;
	}
	else
	{
		lTransId = 1;
	}
	apInfoSet->m_LastTrans.m_NodeId = m_Info.m_Id;
	
	apInfoSet->m_LastTrans.m_TransId = lTransId;
	apInfoSet->m_LastTrans.m_Type = EnumPrepareRequest;

	apInfoSet->m_TransData = std::string(apData, aSize);
	
	for (auto it = m_Parts.begin(); it != m_Parts.end(); it++)
	{
		apInfoSet->m_RelatedNodes.insert(it->first);
	}

	apInfoSet->m_LastTrans.m_Size = sizeof(Trans) + aSize;
	apInfoSet->m_LastTrans.m_Size += (sizeof(uint32_t)*(1+apInfoSet->m_RelatedNodes.size()));

	leveldb::WriteBatch lBatch;

	PC2DataStore::WriteCurrentTransId(&lBatch, lTransId);
	PC2DataStore::WriteTransData(&lBatch, apInfoSet);
	PC2DataStore::WriteLastTransState(&lBatch, apInfoSet);
	PC2DataStore::WriteRelatedNodeSet(&lBatch, apInfoSet);

	m_pStore->CommitBatch(&lBatch);
}




}